package cn.me.alphamq.broker.netty.broker;

import cn.me.alphamq.broker.event.model.broker.*;
import cn.me.alphamq.broker.event.model.cluster.StartSyncEvent;
import cn.me.alphamq.broker.rebalance.ConsumerInstanceManager;
import cn.me.alphamq.common.entity.req.*;
import cn.me.alphamq.common.entity.resp.PushMsgResp;
import cn.me.alphamq.common.entity.resp.StartSyncResp;
import cn.me.alphamq.common.enumeration.EventCodeEnum;
import cn.me.alphamq.common.event.EventBus;
import cn.me.alphamq.common.event.model.Event;
import cn.me.alphamq.common.msg.TcpMsg;
import cn.me.alphamq.common.remote.SyncFuture;
import com.alibaba.fastjson2.JSON;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;

/**
 * broker 的服务器使用的处理器，可共享
 *
 * @author f
 */
@ChannelHandler.Sharable
public class BrokerServerHandler extends SimpleChannelInboundHandler<TcpMsg> {

    private static final Logger log = LoggerFactory.getLogger(BrokerServerHandler.class);

    /**
     * 事件总线
     */
    private final EventBus eventBus;

    public BrokerServerHandler(EventBus eventBus) {
        this.eventBus = eventBus;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TcpMsg tcpMsg) throws Exception {
        int code = tcpMsg.getCode();
        byte[] body = tcpMsg.getBody();

        Event event = null;
        if (EventCodeEnum.PUSH_MSG_REQ.getCode() == code) {
            event = new PushMsgEvent(JSON.parseObject(body, PushMsgReq.class));
        } else if (EventCodeEnum.CONSUME_MSG_REQ.getCode() == code) {
            ConsumeMsgReq consumeMsgReq = JSON.parseObject(body, ConsumeMsgReq.class);
            InetSocketAddress inetSocketAddress = (InetSocketAddress) ctx.channel().remoteAddress();
            consumeMsgReq.setConsumerIp(inetSocketAddress.getHostString());
            consumeMsgReq.setConsumerPort(inetSocketAddress.getPort());
            event = new ConsumeMsgEvent(consumeMsgReq);
        } else if (EventCodeEnum.CONSUME_SUCCESS_REQ.getCode() == code) {
            event = new ConsumeSuccessEvent(JSON.parseObject(body, ConsumeSuccessReq.class));
        } else if (EventCodeEnum.CREATE_TOPIC_REQ.getCode() == code) {
            event = new CreateTopicEvent(JSON.parseObject(body, CreateTopicReq.class));
        } else if (EventCodeEnum.START_SYNC_REQ.getCode() == code) {
            event = new StartSyncEvent(JSON.parseObject(body, StartSyncReq.class));
        } else if (EventCodeEnum.RETRY_CONSUME_MSG_REQ.getCode() == code) {
            event = new RetryConsumeMsgEvent(JSON.parseObject(body, RetryConsumeMsgReq.class));
        } else if (EventCodeEnum.PUSH_DELAY_MSG_REQ.getCode() == code) {
            event = new PushDelayMsgEvent(JSON.parseObject(body, PushDelayMsgReq.class));
        } else if (EventCodeEnum.PUSH_TX_MSG_REQ.getCode() == code) {
            event = new PushTxMsgEvent(JSON.parseObject(body, PushTxMsgReq.class));
        } else if (EventCodeEnum.START_SYNC_RESP.getCode() == code) {
            SyncFuture.removeAndSetResp(body, StartSyncResp.class, tcpMsg);
            return;
        } else if (EventCodeEnum.PUSH_MSG_RESP.getCode() == code) {
            SyncFuture.removeAndSetResp(body, PushMsgResp.class, tcpMsg);
            return;
        } else {
            throw new RuntimeException("Invalid event code: " + code);
        }

        event.setCtx(ctx);
        eventBus.publish(event);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        log.info("new connection built...");
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        log.info("connection closed...");
        // 如果消费者实例断开连接，那么就应该删除它
        InetSocketAddress inetSocketAddress = (InetSocketAddress) ctx.channel().remoteAddress();
        ConsumerInstanceManager.removeInactiveInstance(inetSocketAddress.getHostString(), inetSocketAddress.getPort());
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("exceptionCaught", cause);
    }
}
